package org.apache.ocean.main;

import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.io.Serializable;
import java.util.Date;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.store.RAMDirectory;
import org.apache.ocean.main.TransactionSystem.Category;
import org.apache.ocean.main.util.RamDirectorySerializer;

/**
 * Stays in memory
 * 
 */
public abstract class Batch {
	protected IndexDocuments documents;
	protected RAMDirectory ramDirectory;
	protected Analyzer analyzer;
	protected long schemaVersion;
	protected Deletes deletes;
	protected Date timestamp;
	protected boolean isClosed = false;
	protected String categoryName;
	private IndexSchema.Category schemaCategory;

	public Batch(String categoryName, IndexSchema.Category schemaCategory) {
		this.categoryName = categoryName;
		this.schemaCategory = schemaCategory;
		this.analyzer = schemaCategory.getAnalyzer();
		timestamp = new Date();
	}
  
	public IndexSchema.Category getIndexSchemaCategory() {
		return schemaCategory;
	}
	
	public boolean hasRAMDirectory() {
		return ramDirectory != null;
	}
	
	public RAMDirectory getRamDirectory() {
		return ramDirectory;
	}

	public long getSchemaVersion() {
		return schemaVersion;
	}
	
	public String getCategoryName() {
		return categoryName;
	}
	
	public static class MasterBatch extends Batch {
	  private Category category;
	  
	  public MasterBatch(Category category, IndexSchema.Category schemaCategory) {
	  	super(category.getName(), schemaCategory);
	  	this.category = category;
	  }
	  
	  public Serializable getSerializableRamDirectory(Long id) {
	  	assert ramDirectory != null;
	  	return new SerializableRamDirectory(id, categoryName, ramDirectory);
	  }
	  
	  public Serializable getSerializableBatchDeletes(Long id) {
	  	return new SerializableBatchDeletes(id, categoryName, deletes);
	  }
	  
	  public Serializable getSerializableBatchDocuments(Long id) {
			return new SerializableBatchDocuments(id, categoryName, documents); 
		}
	  
	  public void commit() throws Exception {
			category.commitBatch(this);
		}
	}
	
	public static class SlaveBatch extends Batch {
	  private Long id;
	  
	  public SlaveBatch(String category, Long id, IndexSchema.Category schemaCategory, IndexDocuments documents, Deletes deletes) {
	  	super(category, schemaCategory);
	  	this.id = id;
	  	this.documents = documents;
	  	this.deletes = deletes;
	  }
	  
	  public SlaveBatch(String category, Long id, IndexSchema.Category schemaCategory, RAMDirectory ramDirectory, Deletes deletes) {
	  	super(category, schemaCategory);
	  	this.id = id;
	  	this.ramDirectory = ramDirectory;
	  	this.deletes = deletes;
	  }
	  
	  public Long getId() {
	  	return id;
	  }
	}
	
	public static class SerializableRamDirectory extends SerializableBatch implements Externalizable {
		private static final long serialVersionUID = 1l;
		private RAMDirectory ramDirectory;
		
		public SerializableRamDirectory(Long id, String category, RAMDirectory ramDirectory) {
			super(id, category);
			this.ramDirectory = ramDirectory;
		}
    
		// TODO: use native version not externalizable
		public void readExternal(ObjectInput objectInput) throws IOException, ClassNotFoundException {
			long objectVersion = objectInput.readLong();
			ramDirectory = RamDirectorySerializer.deserialize(objectInput);
		}
		
		public void writeExternal(ObjectOutput objectOutput) throws IOException {
			objectOutput.writeLong(serialVersionUID);
			RamDirectorySerializer.serialize(ramDirectory, objectOutput);
		}
		
		public RAMDirectory getRamDirectory() {
			return ramDirectory;
		}
	}
	
	public static class SerializableBatchDocuments extends SerializableBatch {
		private static final long serialVersionUID = 1l;
		private IndexDocuments documents;
		
		public SerializableBatchDocuments(Long id, String category, IndexDocuments documents) {
			super(id, category);
			this.documents = documents;
		}
		
		public IndexDocuments getDocuments() {
			return documents;
		}
	}
	
	public static class SerializableBatchDeletes extends SerializableBatch {
		private static final long serialVersionUID = 1l;
		private Deletes deletes;
		
		public SerializableBatchDeletes(Long id, String category, Deletes deletes) {
			super(id, category);
		}
		
		public Deletes getDeletes() {
			return deletes;
		}
	}
	
	public abstract static class SerializableBatch implements Serializable {
		private String category;
		private Long id;
		
		public SerializableBatch(Long id, String category) {
			this.id = id;
			this.category = category;
		}
		public String getCategory() {
			return category;
		}
		public Long getId() {
			return id;
		}
	}
	
	public Analyzer getAnalyzer() {
		return analyzer;
	}
	
	public void setRAMDirectory(RAMDirectory ramDirectory) {
		documents = null;
		if (isClosed) throw new RuntimeException("batch closed");
		this.ramDirectory = ramDirectory;
	}
	
	public void setDeletes(Deletes deletes) {
		if (isClosed) throw new RuntimeException("batch closed");
		this.deletes = deletes;
	}
	
	public void addDocuments(IndexDocuments indexDocuments) {
		if (isClosed) throw new RuntimeException("batch closed");
		if (this.documents == null)
			this.documents = new IndexDocuments(indexDocuments);
		this.documents.addAll(indexDocuments);
	}

	public boolean hasDocuments() {
		if (documents == null || documents.size() == 0)
			return false;
		else
			return true;
	}

	public boolean hasDeletes() {
		if (deletes == null || !deletes.hasDeletes()) {
			return false;
		} else {
			return true;
		}
	}

	// disallow any more additions
	public void close() {
		isClosed = true;
	}

	public IndexDocuments getDocuments() {
		return documents;
	}

	public Deletes getDeletes() {
		return deletes;
	}

	public Date getTimestamp() {
		return timestamp;
	}
}
